package main

import (
	"context"
	"fmt"
	"log"
	"sync"
)

type Queue struct {
	name   string
	jobs   chan Job
	ctx    context.Context
	cancel context.CancelFunc
}

type Job struct {
	Name   string
	Action func() error
}

func NewQueue(name string) *Queue {
	ctx, cancel := context.WithCancel(context.Background())

	return &Queue{
		jobs:   make(chan Job),
		name:   name,
		ctx:    ctx,
		cancel: cancel,
	}
}

func (q *Queue) AddJobs(jobs []Job) {
	var wg sync.WaitGroup
	wg.Add(len(jobs))

	for _, job := range jobs {
		go func(job Job) {
			q.AddJob(job)
			wg.Done()
		}(job)
	}
	go func() {
		wg.Wait()
		q.cancel()
	}()
}

func (q *Queue) AddJob(job Job) {
	q.jobs <- job
	log.Printf("New job %s added to %s queue", job.Name, q.name)
}

func (j Job) Run() error {
	log.Printf("job running: %s", j.Name)
	err := j.Action()
	if err != nil {
		return err
	}
	return nil
}

type Worker struct {
	Queue *Queue
}

// NewWorker initializes a new Worker.
func NewWorker(queue *Queue) *Worker {
	return &Worker{
		Queue: queue,
	}
}

// DoWork processes jobs from the queue (jobs channel).
func (w *Worker) DoWork() bool {
	for {
		select {
		// if context was canceled.
		case <-w.Queue.ctx.Done():
			log.Printf("Work done in queue %s: %s!", w.Queue.name, w.Queue.ctx.Err())
			return true
		// if job received.
		case job := <-w.Queue.jobs:
			err := job.Run()
			if err != nil {
				log.Print(err)
				continue
			}
		}
	}
}

// Our products storage.
var products = []string{
	"books",
	"computers",
}

func main() {
	// New products, which we need to add to our products storage.
	newProducts := []string{
		"apples",
		"oranges",
		"wine",
		"bread",
		"orange juice",
	}
	// New queue initialization.
	productsQueue := NewQueue("NewProducts")
	var jobs []Job

	// Range over new products.
	for _, newProduct := range newProducts {
		// We need to do this, because variables declared in for loops are passed by reference.
		// Otherwise, our closure will always receive the last item from the newProducts.
		product := newProduct
		// Defining of the closure, where we add a new product to our simple storage (products slice)
		action := func() error {
			products = append(products, product)
			return nil
		}
		// Append job to jobs slice.
		jobs = append(jobs, Job{
			Name:   fmt.Sprintf("Importing new product: %s", newProduct),
			Action: action,
		})
	}

	// Adds jobs to the queue.
	productsQueue.AddJobs(jobs)

	// Defines a queue worker, which will execute our queue.
	worker := NewWorker(productsQueue)
	// Execute jobs in queue.
	worker.DoWork()

	// Prints products storage after queue execution.
	defer fmt.Print(products)
}
